SS-455 - source: commit Kafka offsets on change and on a timer by martykulma · Pull Request #39462 · MaterializeInc/materialize

martykulma · 2026-10-01T18:35:23Z

Each worker commits only when its offsets change, plus on a timer. The timer is needed because the consumer only ever assigns partitions, which results in the brokers treating the group as standalone. For standalone groups, brokers expire a partition's offset offsets.retention.minutes after its last commit, whether or not it is still current. The interval is the kafka_offset_commit_refresh_interval dyncfg: 10 minutes by default, 10 seconds in CI, zero disables it. A failed commit is retried on the next resume upper.

🤖 Generated with Claude Code
👨 Improved by human

linear-code · 2026-10-01T18:35:52Z

SS-455

def- · 2026-10-01T20:33:36Z

QA LLM Review

1. MEDIUM -- Idle partitions are never re-committed, so the broker expires their consumer-group offsets

src/storage/src/source/kafka.rs:1163

With the offsets != self.committed_offsets check, a worker whose partitions receive no new messages commits them once and never again, even while the rest of the source keeps advancing. Materialize consumes with assign(), which makes it a standalone consumer group, and brokers delete a standalone group's offset for a partition offsets.retention.minutes (7 days by default) after that partition's last commit. Lag-monitoring tools, which the Kafka source docs say these commits exist for, then lose those partitions.

Details

On main, reclock_committed_upper calls tx.send_replace every time it runs (source_reader_pipeline.rs:685 at the merge base), so every worker re-commits its offsets about once per tick, and that refreshes the commit timestamp. On #39461, any change to ResumeUppers still re-commits every worker's partitions. This PR takes away that last refresh for idle partitions. Two examples: a keyed topic where some partitions get no writes for a week, or a cluster with at least as many workers as partitions, where each worker owns at most one partition. After the retention window, kafka-consumer-groups --describe shows - for those partitions, and lag exporters either drop them or report them as missing. Data ingestion is not affected.

Suggested fix: keep the dedup, but re-commit unchanged offsets once the last successful commit is older than a refresh interval set well below typical broker retention (for example one hour, or a dyncfg). Because #39461 switched to send_if_modified, process_frontier is never called for a fully idle source. So the refresh needs a timer in resume_uppers_process_loop, for example a select! over the stream and a tokio::time::interval, rather than only a looser condition here. The same timer would also retry a failed commit on an idle source. Today the doc comment's "retried on the next resume upper" may never happen for such a source.

martykulma · 2026-10-02T14:24:21Z

MEDIUM -- Idle partitions are never re-committed, so the broker expires their consumer-group offsets

Confirmed and addressed

def- · 2026-10-05T13:07:13Z

QA LLM Review

1. MEDIUM -- `"10m"` is not a valid system-parameter duration, so environmentd fails to boot in random-parameter CI runs

misc/python/materialize/mzcompose/__init__.py:428

System variables parse durations with the units us, ms, s, min, h and d. There is no m, so "10m" is rejected. Under CI_SYSTEM_PARAMETERS=random, about one job in three picks "10m", and every Materialize container in that job then panics at startup. All tests in those jobs fail, which hides whatever the random-parameter run was meant to find.

Details

Dyncfgs that become SQL system vars are parsed by impl Value for Duration (src/sql/src/session/vars/value.rs:307), not by humantime, and that parser has no m unit. get_default_system_parameters passes the chosen value as a system-parameter default. When catalog open hits a parse error it returns that error instead of skipping the parameter (src/adapter/src/catalog/open.rs:263).

I reproduced this on a current PR image using an existing duration dyncfg:

MZ_SYSTEM_PARAMETER_DEFAULT="kafka_poll_max_wait=10m"
environmentd: fatal: parameter "kafka_poll_max_wait" cannot have value "m": expected us, ms, s, min, h, or d but got "m"

Every other duration in this file already uses min (for example "1min" at line 578). Changing the value to "10min" fixes it.

patrickwwbutler

Small question, plus suggestion to fix the LLM lint, but no blockers! LGTM

patrickwwbutler · 2026-10-05T17:48:02Z

+        # Low default so CI exercises the periodic recommit, which production
+        # only reaches after ten minutes.
+        VariableSystemParameter(
+            "kafka_offset_commit_refresh_interval", "10s", ["1s", "10s", "10m"]


Suggested change

"kafka_offset_commit_refresh_interval", "10s", ["1s", "10s", "10m"]

"kafka_offset_commit_refresh_interval", "10s", ["1s", "10s", "600s"]

patrickwwbutler · 2026-10-05T17:49:51Z

+                // Zero disables the refresh. `interval` panics on a zero period, so the tick arm
+                // below is what honors it.
+                let mut refresh =
+                    tokio::time::interval(refresh_interval.max(Duration::from_millis(1)));


if the value was set to 0, and this clamps it to 1ms, is that sustainable? I suppose it would just happen every iteration of the loop, which is okay?

1ms would just update as fast as the calls would allow it (because the missed tick behavior is delay).

If someone sets it to 0 to disable, then the branch never fires because the precondition would return false:

materialize/src/storage/src/source/kafka.rs

Lines 732 to 735 in 97b4ca9

_ = refresh.tick(), if !refresh_interval.is_zero() => {

if let Err(e) = offset_committer.refresh().await {

report(e, &"refresh");

}

https://docs.rs/tokio/latest/tokio/macro.select.html

Evaluate all provided expressions. If the precondition returns false, disable the branch for the remainder of the current call to select!. Re-entering select! due to a loop clears the “disabled” state.

Brokers expire a standalone group's offset per partition offsets.retention.minutes after its last commit, so offsets the dedupe skips must still be recommitted periodically. The interval is the kafka_offset_commit_refresh_interval dyncfg, 10 minutes by default and 10 seconds in CI, with zero disabling the refresh. Co-Authored-By: Claude <noreply@anthropic.com>

martykulma added this pull request to stack #39464 October 1, 2026 18:35

martykulma changed the title ~~source: Kafka only comits offsets for a partition if they've changed~~ SS-455 - source: Kafka only commits offsets for a partition if they've changed Oct 1, 2026

martykulma marked this pull request as ready for review October 1, 2026 19:30

martykulma requested a review from a team as a code owner October 1, 2026 19:30

martykulma requested a review from patrickwwbutler October 1, 2026 19:31

martykulma force-pushed the maz-kafka-offset-commit-dedupe branch from 7e97474 to e1893ed Compare October 2, 2026 14:21

martykulma requested a review from a team as a code owner October 2, 2026 14:21

martykulma changed the title ~~SS-455 - source: Kafka only commits offsets for a partition if they've changed~~ SS-455 - source: commit Kafka offsets on change and on a timer Oct 2, 2026

patrickwwbutler approved these changes Oct 5, 2026

View reviewed changes

martykulma force-pushed the maz-kafka-offset-commit-dedupe branch from e1893ed to 97b4ca9 Compare October 5, 2026 17:56

martykulma and others added 2 commits October 5, 2026 14:13

source: Kafka only comits offsets for a partition if they've changed

dca72ce

martykulma force-pushed the maz-kafka-offset-commit-dedupe branch from 97b4ca9 to c220506 Compare October 5, 2026 18:13

martykulma merged commit 491235b into main Oct 5, 2026
84 checks passed

martykulma deleted the maz-kafka-offset-commit-dedupe branch October 5, 2026 18:44

Navigation Menu

Repository navigation

SS-455 - source: commit Kafka offsets on change and on a timer - #39462

SS-455 - source: commit Kafka offsets on change and on a timer#39462
martykulma merged 2 commits into
maz-offset-committed-per-exportfrom
maz-kafka-offset-commit-dedupe

martykulma commented Oct 1, 2026 •

edited

Loading

Uh oh!

linear-code Bot commented Oct 1, 2026

Uh oh!

def- commented Oct 1, 2026

Uh oh!

martykulma commented Oct 2, 2026

Uh oh!

def- commented Oct 5, 2026

Uh oh!

patrickwwbutler left a comment

Uh oh!

patrickwwbutler Oct 5, 2026

Uh oh!

patrickwwbutler Oct 5, 2026

Uh oh!

martykulma Oct 5, 2026

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

	"kafka_offset_commit_refresh_interval", "10s", ["1s", "10s", "10m"]
	"kafka_offset_commit_refresh_interval", "10s", ["1s", "10s", "600s"]

	_ = refresh.tick(), if !refresh_interval.is_zero() => {
	if let Err(e) = offset_committer.refresh().await {
	report(e, &"refresh");
	}

Conversation

martykulma commented Oct 1, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

linear-code Bot commented Oct 1, 2026

Uh oh!

def- commented Oct 1, 2026

QA LLM Review

1. MEDIUM -- Idle partitions are never re-committed, so the broker expires their consumer-group offsets

Uh oh!

martykulma commented Oct 2, 2026

Uh oh!

def- commented Oct 5, 2026

QA LLM Review

1. MEDIUM -- "10m" is not a valid system-parameter duration, so environmentd fails to boot in random-parameter CI runs

Uh oh!

patrickwwbutler left a comment

Choose a reason for hiding this comment

Uh oh!

patrickwwbutler Oct 5, 2026

Choose a reason for hiding this comment

Uh oh!

patrickwwbutler Oct 5, 2026

Choose a reason for hiding this comment

Uh oh!

martykulma Oct 5, 2026

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

martykulma commented Oct 1, 2026 •

edited

Loading

1. MEDIUM -- `"10m"` is not a valid system-parameter duration, so environmentd fails to boot in random-parameter CI runs